package xyz.zcq.zcqrpc.registry;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Watch;
import io.etcd.jetcd.kv.DeleteResponse;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;
import xyz.zcq.zcqrpc.config.RegistryConfig;
import xyz.zcq.zcqrpc.model.ServiceMetaInfo;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * Etcd注册中心
 */
@Slf4j
public class EtcdRegistry implements Registry {

    /**
     * 根节点路径
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";
    /**
     * 客户端
     */
    private Client client;
    /**
     * 本机注册的节点key集合（用于维护续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();
    /**
     * 注册中心服务缓存
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();
    /**
     * 正在监听的服务节点key的集合
     */
    private final Set<String> watchKeySet = new ConcurrentHashSet<>();

    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder().endpoints(registryConfig.getAddress()).connectTimeout(Duration.ofMillis(registryConfig.getTimeout())).build();
        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        String key = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence keyBs = ByteSequence.from(key, StandardCharsets.UTF_8);
        ByteSequence valueBs = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);
        PutOption putOption = PutOption.builder().withLeaseId(client.getLeaseClient().grant(30).get().getID()).build();

        PutResponse putResponse = client.getKVClient().put(keyBs, valueBs, putOption).get();
        localRegisterNodeKeySet.add(key);
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) throws Exception {
        String key = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence keyBs = ByteSequence.from(key, StandardCharsets.UTF_8);
        DeleteResponse deleteResponse = client.getKVClient().delete(keyBs).get();
        localRegisterNodeKeySet.remove(key);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        List<ServiceMetaInfo> cache = this.registryServiceCache.getCache(serviceKey);
        if (CollectionUtil.isNotEmpty(cache)) {
            log.info("从缓存中获取服务元信息，cache: {}", JSONUtil.toJsonStr(cache));
            return cache;
        }

        ByteSequence keyPrefixBs = ByteSequence.from(ETCD_ROOT_PATH + serviceKey + "/", StandardCharsets.UTF_8);
        GetOption getOption = GetOption.builder().isPrefix(true).build();

        try {
            List<KeyValue> keyValueList = client.getKVClient().get(keyPrefixBs, getOption).get().getKvs();
            List<ServiceMetaInfo> serviceMetaInfoList = keyValueList.stream().map(e -> {
                ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(e.getValue().toString(StandardCharsets.UTF_8), ServiceMetaInfo.class);
                watch(serviceMetaInfo);
                return serviceMetaInfo;
            }).collect(Collectors.toList());
            log.info("从注册中心获取服务元信息, serviceMetaInfoList: {}", JSONUtil.toJsonStr(serviceMetaInfoList));
            this.registryServiceCache.writeCache(serviceKey, serviceMetaInfoList);
            return serviceMetaInfoList;

        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败", e);
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");

        // 下线注册中心的节点
        for (String key : localRegisterNodeKeySet) {
            try {
                client.getKVClient().delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        // 关闭资源
        if (client != null) {
            client.close();
        }
    }

    @Override
    public void heartBeat() {
        CronUtil.schedule("*/10 * * * * *", (Task) () -> {
            for (String key : localRegisterNodeKeySet) {
                try {
                    ByteSequence keyBs = ByteSequence.from(key, StandardCharsets.UTF_8);
                    List<KeyValue> kvs = client.getKVClient().get(keyBs).get().getKvs();
                    if (CollectionUtil.isEmpty(kvs)) {
                        continue;
                    }
                    KeyValue keyValue = kvs.get(0);
                    ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(keyValue.getValue().toString(StandardCharsets.UTF_8), ServiceMetaInfo.class);
                    register(serviceMetaInfo);
                } catch (Exception e) {
                    throw new RuntimeException(key + "续签失败", e);
                }
            }
        });
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(ServiceMetaInfo serviceMetaInfo) {
        String key = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        boolean addSuc = this.watchKeySet.add(key);
        if (addSuc) {
            Watch watch = client.getWatchClient();
            watch.watch(ByteSequence.from(key, StandardCharsets.UTF_8), watchResponse -> {
                List<WatchEvent> events = watchResponse.getEvents();
                for (WatchEvent event : events) {
                    switch (event.getEventType()) {
                        case DELETE:
                            this.registryServiceCache.clearCache(serviceMetaInfo.getServiceKey());
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }

}
